package com.share.example.json2flat.service.impl;

import com.alibaba.fastjson.JSON;
import com.baomidou.dynamic.datasource.annotation.DS;
import com.share.common.enums.DbTypeEnum;
import com.share.common.utils.DbUtil;
import com.share.config.redisson.RedisLock;
import com.share.example.json2flat.constants.ConfConstant;
import com.share.example.json2flat.service.HbaseService;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.phoenix.schema.ColumnNotFoundException;
import org.apache.phoenix.schema.TableNotFoundException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils;

import javax.sql.DataSource;
import java.sql.Connection;
import java.sql.SQLException;
import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

@DS("hbase")
@Slf4j
@Service
public class HbaseServiceImpl implements HbaseService {
    @Autowired
    private RedisLock redisLock;
    @Autowired
    private DataSource dataSource;

    @Override
    public Integer save(Object value) throws Exception {
        int re = 0;
        String strValue = value.toString();
        if (StringUtils.isNotBlank(strValue)) {
            Map<String, Object> map = JSON.parseObject(strValue, Map.class);
            String tableName = (String) map.get(ConfConstant.TABLE_NAME);
            map.remove(ConfConstant.TABLE_NAME);
            map.put("id", "NEXT VALUE FOR " + "id_seq_" + tableName);
            String insert = DbUtil.toInsert(tableName, map, DbTypeEnum.PHOENIX.code);
            Connection con = dataSource.getConnection();
            try {
                Long aLong = DbUtil.insertOneRecord(con, insert, false);
                if (aLong > 0) {
                    re = 1;
                }
            } catch (TableNotFoundException e1) {//建表
                log.error("表不存在,即将触发建表tableName={}", tableName);
                String LOCK_KEY_CREATE_TABLE_LOCK = "create_table_lock";
                boolean locked = false;
                //建表
                try {
                    locked = redisLock.tryLock(LOCK_KEY_CREATE_TABLE_LOCK, 3, TimeUnit.SECONDS);
                    if (locked) {
                        log.info("即将执行建表,tableName={}", tableName);
                        createTable(con, tableName, map);
                    }
                } catch (Exception e11) {
                    log.error("未获取到锁,补偿创建表异常tableName={}", tableName, e11);
                } finally {
                    if (locked) {
                        redisLock.unlock(LOCK_KEY_CREATE_TABLE_LOCK);
                    }
                }

                //重新插入
                Long aLong = DbUtil.insertOneRecord(con, insert, false);
                if (aLong > 0) {
                    re = 1;
                }
            } catch (ColumnNotFoundException e2) {//添加列
                //需要使用分布式锁-并发问题需要处理
                log.error("列不存在,即将触发建表tableName={},columnName={}", tableName, e2.getColumnName());
                String LOCK_KEY_ADD_COLUMN_LOCK = "add_column_lock";
                boolean locked = false;
                try {
                    locked = redisLock.tryLock(LOCK_KEY_ADD_COLUMN_LOCK, 10, TimeUnit.SECONDS);
                    if (locked) {
                        //org.apache.phoenix.schema.ColumnNotFoundException: ERROR 504 (42703): Undefined column. columnName=USER_INFO2.USERID1
                        List<String> list = DbUtil.showColumns(con, tableName, DbTypeEnum.PHOENIX.code);
                        List<String> paramKey = map.keySet().stream().map(t -> t.toUpperCase()).collect(Collectors.toList());
                        paramKey.removeAll(list);

                        for (String columnName : paramKey) {//新增列
                            addColumn(con, tableName, columnName);
                        }

                        log.error("列创建完毕,tableName={},columnName={}", tableName, e2.getColumnName());
                    }
                } catch (Exception e21) {
                    log.error("补偿创建列异常tableName={},columnName={}", tableName, e2.getColumnName(), e21);
                } finally {
                    if (locked) {
                        redisLock.unlock(LOCK_KEY_ADD_COLUMN_LOCK);
                    }
                }

                //重新插入
                Long aLong = DbUtil.insertOneRecord(con, insert, false);
                if (aLong > 0) {
                    re = 1;
                }
            } catch (Exception e3) {
                throw e3;
            } finally {
                if (con != null) {
                    con.close();
                }
            }
        }
        return re;
    }

    @Override
    public Map<String, Object> deleteLog(Integer retainNum) throws Exception {
        Map<String, Object> reMap = null;
        Connection connection = null;
        try {
            connection = dataSource.getConnection();
            List<String> tables = DbUtil.showTables(connection, DbTypeEnum.PHOENIX.code);
            if (!CollectionUtils.isEmpty(tables)) {
                String[] flat = tables.stream().filter(e -> e.toUpperCase().startsWith("FJ_")).collect(Collectors.toList()).toArray(new String[0]);
                reMap = DbUtil.deleteDataById(connection, flat, retainNum);
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            if (connection != null) {
                connection.close();
            }
        }

        return reMap;
    }

    //新增一列
    private void addColumn(Connection con, String tableName, String columnName) throws SQLException {
        //创建表结构
        String addColumnSql = "ALTER TABLE " + tableName + " ADD IF NOT EXISTS " + columnName + " VARCHAR";
        DbUtil.deleteOrUpdate(con, addColumnSql);
    }

    //创建表结构
    private void createTable(Connection con, String tableName, Object map) throws SQLException {
        //创建序列
        String seqSql = "CREATE SEQUENCE IF NOT EXISTS id_seq_" + tableName + " START 1 CACHE 100";
        DbUtil.deleteOrUpdate(con, seqSql);

        //创建表
        String createTableSql = DbUtil.toHbaseCreateTable(tableName, map);
        DbUtil.deleteOrUpdate(con, createTableSql);
    }
}
